package com.gxlevi.kafka.stream;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStreamBuilder;

import java.util.Properties;

/**
 * 流式的编程
 * 通过streamAPI实现将数据从test里面读取出来,写入到test2里面去
 */
public class Stream {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG,"bigger");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"node01:9092");
        props.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());//key的序列化和反序列化的类
        props.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG,Serdes.String().getClass());//value的序列化和反序列化的类
        //获取核心类 KStreamBuilder
        KStreamBuilder kStreamBuilder = new KStreamBuilder();
        //通过KStreamBuilder调用stream方法表示从哪个topic当中获取数据
        kStreamBuilder.stream("test")
                .mapValues(line -> line.toString().toUpperCase())//表示将每一行value都取出来
                .to("test2");//将数据写入test2里面去
        //通过kStreamBuilder可以用于创建KafkaStream 通过kafkaStream来实现流式的编程启动
        KafkaStreams kafkaStreams = new KafkaStreams(kStreamBuilder, props);
        kafkaStreams.start();//调用start启动kafka的流API
    }
}
